"""
幂等性管理器
确保数据同步操作的幂等性，防止重复处理相同数据
"""

import hashlib
import json
import time
from datetime import datetime
from typing import Dict, Any, Optional, Set, List
from dataclasses import dataclass
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, insert, update, delete, text
import logging

logger = logging.getLogger(__name__)


def json_serializer(obj):
    """JSON序列化器，处理datetime、Decimal等特殊对象"""
    from decimal import Decimal

    if isinstance(obj, datetime):
        return obj.strftime('%Y-%m-%d %H:%M:%S')
    elif isinstance(obj, Decimal):
        return str(obj)
    raise TypeError(f"Object of type {type(obj)} is not JSON serializable")


def safe_json_dumps(data, **kwargs):
    """安全的JSON序列化，处理datetime对象"""
    return json.dumps(data, default=json_serializer, ensure_ascii=False, **kwargs)


@dataclass
class IdempotencyRecord:
    """幂等性记录"""
    key: str
    task_id: int
    execution_id: int
    data_hash: str
    processed_at: float
    batch_number: Optional[int] = None
    record_count: int = 0
    metadata: Optional[Dict[str, Any]] = None


class IdempotencyManager:
    """幂等性管理器"""
    
    def __init__(self, db: AsyncSession):
        """
        初始化幂等性管理器
        
        Args:
            db: 数据库会话
        """
        self.db = db
        self._memory_cache: Dict[str, IdempotencyRecord] = {}
        self._cache_ttl = 3600  # 1小时缓存TTL
    
    def _generate_data_hash(self, data: List[Dict[str, Any]]) -> str:
        """
        生成数据哈希值

        Args:
            data: 数据列表

        Returns:
            数据的MD5哈希值
        """
        # 对数据进行排序和序列化，确保相同数据产生相同哈希
        sorted_data = sorted(data, key=lambda x: safe_json_dumps(x, sort_keys=True))
        data_str = safe_json_dumps(sorted_data, sort_keys=True)
        return hashlib.md5(data_str.encode('utf-8')).hexdigest()
    
    def _generate_idempotency_key(
        self, 
        task_id: int, 
        execution_id: int, 
        batch_number: Optional[int] = None,
        additional_key: Optional[str] = None
    ) -> str:
        """
        生成幂等性键
        
        Args:
            task_id: 任务ID
            execution_id: 执行ID
            batch_number: 批次号
            additional_key: 额外的键值
            
        Returns:
            幂等性键
        """
        key_parts = [str(task_id), str(execution_id)]
        
        if batch_number is not None:
            key_parts.append(f"batch_{batch_number}")
        
        if additional_key:
            key_parts.append(additional_key)
        
        return ":".join(key_parts)
    
    async def _get_from_database(self, key: str) -> Optional[IdempotencyRecord]:
        """从数据库获取幂等性记录"""
        try:
            # 这里需要创建幂等性记录表，暂时使用内存缓存
            # 实际实现中应该查询数据库表
            return None
        except Exception as e:
            logger.error(f"从数据库获取幂等性记录失败: {e}")
            return None
    
    async def _save_to_database(self, record: IdempotencyRecord):
        """保存幂等性记录到数据库"""
        try:
            # 这里需要创建幂等性记录表，暂时使用内存缓存
            # 实际实现中应该插入到数据库表
            pass
        except Exception as e:
            logger.error(f"保存幂等性记录到数据库失败: {e}")
    
    async def check_and_record(
        self,
        task_id: int,
        execution_id: int,
        data: List[Dict[str, Any]],
        batch_number: Optional[int] = None,
        additional_key: Optional[str] = None
    ) -> bool:
        """
        检查并记录幂等性
        
        Args:
            task_id: 任务ID
            execution_id: 执行ID
            data: 数据列表
            batch_number: 批次号
            additional_key: 额外的键值
            
        Returns:
            True表示数据未处理过，False表示数据已处理过
        """
        if not data:
            return True
        
        # 生成幂等性键和数据哈希
        idempotency_key = self._generate_idempotency_key(
            task_id, execution_id, batch_number, additional_key
        )
        data_hash = self._generate_data_hash(data)
        
        # 先检查内存缓存
        if idempotency_key in self._memory_cache:
            cached_record = self._memory_cache[idempotency_key]
            
            # 检查缓存是否过期
            if time.time() - cached_record.processed_at < self._cache_ttl:
                if cached_record.data_hash == data_hash:
                    logger.debug(f"数据已处理过（内存缓存）: {idempotency_key}")
                    return False
            else:
                # 缓存过期，删除
                del self._memory_cache[idempotency_key]
        
        # 检查数据库
        db_record = await self._get_from_database(idempotency_key)
        if db_record and db_record.data_hash == data_hash:
            logger.debug(f"数据已处理过（数据库）: {idempotency_key}")
            
            # 更新内存缓存
            self._memory_cache[idempotency_key] = db_record
            return False
        
        # 数据未处理过，记录幂等性信息
        record = IdempotencyRecord(
            key=idempotency_key,
            task_id=task_id,
            execution_id=execution_id,
            data_hash=data_hash,
            processed_at=time.time(),
            batch_number=batch_number,
            record_count=len(data),
            metadata={
                'additional_key': additional_key,
                'first_record_sample': data[0] if data else None
            }
        )
        
        # 保存到内存缓存和数据库
        self._memory_cache[idempotency_key] = record
        await self._save_to_database(record)
        
        logger.debug(f"记录新的幂等性信息: {idempotency_key}")
        return True
    
    async def is_processed(
        self,
        task_id: int,
        execution_id: int,
        data: List[Dict[str, Any]],
        batch_number: Optional[int] = None,
        additional_key: Optional[str] = None
    ) -> bool:
        """
        检查数据是否已处理过
        
        Args:
            task_id: 任务ID
            execution_id: 执行ID
            data: 数据列表
            batch_number: 批次号
            additional_key: 额外的键值
            
        Returns:
            True表示已处理过，False表示未处理过
        """
        return not await self.check_and_record(
            task_id, execution_id, data, batch_number, additional_key
        )
    
    async def mark_as_processed(
        self,
        task_id: int,
        execution_id: int,
        data: List[Dict[str, Any]],
        batch_number: Optional[int] = None,
        additional_key: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None
    ):
        """
        标记数据为已处理
        
        Args:
            task_id: 任务ID
            execution_id: 执行ID
            data: 数据列表
            batch_number: 批次号
            additional_key: 额外的键值
            metadata: 元数据
        """
        if not data:
            return
        
        idempotency_key = self._generate_idempotency_key(
            task_id, execution_id, batch_number, additional_key
        )
        data_hash = self._generate_data_hash(data)
        
        record = IdempotencyRecord(
            key=idempotency_key,
            task_id=task_id,
            execution_id=execution_id,
            data_hash=data_hash,
            processed_at=time.time(),
            batch_number=batch_number,
            record_count=len(data),
            metadata=metadata
        )
        
        # 保存到内存缓存和数据库
        self._memory_cache[idempotency_key] = record
        await self._save_to_database(record)
        
        logger.debug(f"标记数据为已处理: {idempotency_key}")
    
    async def clear_execution_records(self, execution_id: int):
        """
        清理指定执行的幂等性记录
        
        Args:
            execution_id: 执行ID
        """
        # 清理内存缓存
        keys_to_remove = [
            key for key, record in self._memory_cache.items()
            if record.execution_id == execution_id
        ]
        
        for key in keys_to_remove:
            del self._memory_cache[key]
        
        # 清理数据库记录
        try:
            # 这里需要删除数据库中的记录
            pass
        except Exception as e:
            logger.error(f"清理数据库幂等性记录失败: {e}")
        
        logger.info(f"清理执行 {execution_id} 的幂等性记录")
    
    async def clear_expired_records(self, ttl_seconds: int = 86400):
        """
        清理过期的幂等性记录
        
        Args:
            ttl_seconds: 过期时间（秒），默认24小时
        """
        current_time = time.time()
        expired_keys = []
        
        # 清理内存缓存中的过期记录
        for key, record in self._memory_cache.items():
            if current_time - record.processed_at > ttl_seconds:
                expired_keys.append(key)
        
        for key in expired_keys:
            del self._memory_cache[key]
        
        # 清理数据库中的过期记录
        try:
            # 这里需要删除数据库中的过期记录
            pass
        except Exception as e:
            logger.error(f"清理数据库过期记录失败: {e}")
        
        if expired_keys:
            logger.info(f"清理了 {len(expired_keys)} 条过期的幂等性记录")
    
    def get_cache_stats(self) -> Dict[str, Any]:
        """获取缓存统计信息"""
        current_time = time.time()
        
        total_records = len(self._memory_cache)
        expired_records = sum(
            1 for record in self._memory_cache.values()
            if current_time - record.processed_at > self._cache_ttl
        )
        
        return {
            'total_records': total_records,
            'active_records': total_records - expired_records,
            'expired_records': expired_records,
            'cache_ttl': self._cache_ttl,
            'memory_usage_mb': sum(
                len(str(record)) for record in self._memory_cache.values()
            ) / 1024 / 1024
        }


# 创建幂等性记录表的SQL（需要在数据库初始化时执行）
CREATE_IDEMPOTENCY_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS dp_sync_idempotency (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    idempotency_key VARCHAR(255) NOT NULL UNIQUE,
    task_id BIGINT NOT NULL,
    execution_id BIGINT NOT NULL,
    data_hash VARCHAR(32) NOT NULL,
    batch_number INT,
    record_count INT DEFAULT 0,
    metadata JSON,
    processed_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    
    INDEX idx_task_execution (task_id, execution_id),
    INDEX idx_processed_at (processed_at),
    
    FOREIGN KEY (task_id) REFERENCES dp_sync_tasks(id) ON DELETE CASCADE,
    FOREIGN KEY (execution_id) REFERENCES dp_sync_executions(id) ON DELETE CASCADE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='数据同步幂等性记录表';
"""
